package groupOne.app.DWS;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import groupOne.app.BaseAppOneStream;
import groupOne.bean.TrafficPageViewBean;
import groupOne.common.Constant;
import groupOne.util.AtguiguUtil;
import groupOne.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.List;
import java.util.Map;

public class Dws_userTraffic extends BaseAppOneStream {
    public static void main(String[] args) {
        new Dws_userTraffic().init(5004,2,"Dws_userTraffic", Constant.TOPIC_DWD_TRAFFIC_PAGE);
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        //统计一个窗口中的会话数量和页面浏览数量 跳出数量 独立访客数量
        /*
        * {
    "common": {
        "sc": "1",
        "ar": "34",
        "uid": "956",
        "os": "iOS 13.2.9",
        "ch": "Appstore",
        "is_new": "0",
        "md": "iPhone 8",
        "mid": "mid_318",
        "vc": "v2.1.134",
        "ba": "iPhone",
        "sid": "47e23650-49ed-4d56-b01b-41b90c4e607b"
    },
    "page": {
        "page_id": "course_detail",
        "item": "434",
        "during_time": 14385,
        "item_type": "course_id",
        "last_page_id": "mine"
    },
    "ts": 1645552545655
}
        *
        * */
        //1.将数据转化为json格式
        SingleOutputStreamOperator<JSONObject> jsonStream = stream.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector){
                try {
                    JSONObject jsonObject = JSON.parseObject(s);
                    collector.collect(jsonObject);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner((a, b) -> a.getLong("ts")));

        //匹配跳出数据
        Pattern<JSONObject, JSONObject> pattern = Pattern.<JSONObject>begin("first")
                .where(new SimpleCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject jsonObject){
                        return jsonObject.getJSONObject("page").getString("last_page_id") == null;
                    }
                })
                .next("second")
                .where(new SimpleCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject jsonObject){
                        return jsonObject.getJSONObject("page").getString("last_page_id") == null;
                    }
                }).within(Time.seconds(5));

        KeyedStream<JSONObject, String> keyedStream = jsonStream.keyBy(a -> a.getJSONObject("common").getString("uid"));

        SingleOutputStreamOperator<TrafficPageViewBean> patternStream = CEP.pattern(keyedStream, pattern).select(new OutputTag<TrafficPageViewBean>("TrafficPageViewBean") {},
                new PatternTimeoutFunction<JSONObject, TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean timeout(Map<String, List<JSONObject>> map, long l){
                        JSONObject first = map.get("first").get(0);
                        JSONObject common = first.getJSONObject("common");
                        Long ts = first.getLong("ts");
                        String date = AtguiguUtil.toDate(ts);

                        TrafficPageViewBean trafficPageViewBean = new TrafficPageViewBean("","",common.getString("uid"),
                                date,0L,0L,1L,0L,
                                1L,ts);
                        return trafficPageViewBean;
                    }
                },
                new PatternSelectFunction<JSONObject, TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean select(Map<String, List<JSONObject>> map){
                        JSONObject first = map.get("first").get(0);
                        JSONObject common = first.getJSONObject("common");
                        Long ts = first.getLong("ts");
                        String date = AtguiguUtil.toDate(ts);

                        TrafficPageViewBean trafficPageViewBean = new TrafficPageViewBean("","",common.getString("uid"),
                                date,0L,0L,1L,0L,
                                1L,ts);
                        return trafficPageViewBean;
                    }
                }
        );

        DataStream<TrafficPageViewBean> sideOutput = patternStream.getSideOutput(new OutputTag<TrafficPageViewBean>("TrafficPageViewBean") {});

        //2.统计 独立访客数
        //会话总数
        //会话平均浏览页面数
        //会话平均停留时长
        jsonStream
                .map(new MapFunction<JSONObject, TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean map(JSONObject jsonObject){
                    JSONObject common = jsonObject.getJSONObject("common");
                    JSONObject page = jsonObject.getJSONObject("page");
                    Long ts = jsonObject.getLong("ts");
                    String date = AtguiguUtil.toDate(ts);

                    TrafficPageViewBean trafficPageViewBean = new TrafficPageViewBean("","",common.getString("uid"),
                            date,0L,page.getString("last_page_id") == null ? 1L : 0L,1L,page.getLong("during_time"),
                            0L,ts);
                return trafficPageViewBean;
            }
        })
                //合流
                .union(patternStream,sideOutput)
                .keyBy(TrafficPageViewBean::getUid)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce((ReduceFunction<TrafficPageViewBean>) (t1, t2) -> {
                    //设置会话数
                    t1.setSvCt(t1.getSvCt() + t2.getSvCt());
                    //设置页面浏览数
                    t1.setPvCt(t1.getPvCt() + t2.getPvCt());
                    //累积访问时长
                    t1.setDurSum(t1.getDurSum() + t2.getDurSum());
                    return t1;
                }, new ProcessWindowFunction<TrafficPageViewBean, TrafficPageViewBean, String, TimeWindow>() {

                    String curDate;
                    ValueState<String> date;

                    @Override
                    public void open(Configuration parameters){
                        date = getRuntimeContext().getState(new ValueStateDescriptor<>("date", String.class));
                    }

                    @Override
                    public void process(String s, Context context, Iterable<TrafficPageViewBean> iterable, Collector<TrafficPageViewBean> collector) throws Exception {
                        TrafficPageViewBean viewBean = iterable.iterator().next();
                        //状态判断当天的第一个窗口
                        curDate = viewBean.getCurDate();
                        if(!curDate.equals(date.value())){
                            viewBean.setUvCt(1L);
                            date.update(curDate);
                        }
                        //设置窗口时间
                        String stt = AtguiguUtil.toDateTime(context.window().getStart());
                        String edt = AtguiguUtil.toDateTime(context.window().getEnd());
                        viewBean.setStt(stt);
                        viewBean.setEdt(edt);

                        collector.collect(viewBean);
                    }
                })
                .keyBy(TrafficPageViewBean::getStt)
                .reduce((t1,t2) -> {
                    //设置会话数
                    t1.setSvCt(t1.getSvCt() + t2.getSvCt());
                    //设置页面浏览数
                    t1.setPvCt(t1.getPvCt() + t2.getPvCt());
                    //累积访问时长
                    t1.setDurSum(t1.getDurSum() + t2.getDurSum());
                    //设置跳出会话数
                    t1.setUjCt(t1.getUjCt()+t2.getUjCt());
                    return t1;
                })
                .map(bean -> {
                    SerializeConfig config = new SerializeConfig();
                    config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;  // 转成json的时候, 属性名使用下划线
                    return JSON.toJSONString(bean, config);}
                )
                .addSink(FlinkSinkUtil.getDorisSink("edu.dws_traffic_page_view_window"));

/*
drop table if exists dws_traffic_page_view_window;
create table if not exists dws_traffic_page_view_window
(
    `stt`      DATETIME comment '窗口起始时间',
    `edt`      DATETIME comment '窗口结束时间',
    `cur_date` DATE comment '当天日期',
    `uv_ct`    BIGINT replace comment '独立访客数',
    `sv_ct`    BIGINT replace comment '会话数',
    `pv_ct`    BIGINT replace comment '页面浏览数',
    `dur_sum`  BIGINT replace comment '页面访问时长',
    `uj_ct`    BIGINT replace comment '跳出会话数'
) engine = olap aggregate key (`stt`, `edt`, `cur_date`)
comment "流量域版本-渠道-地区-访客类别粒度页面浏览各窗口汇总表"
partition by range(`cur_date`)()
distributed by hash(`stt`, `edt`, `cur_date`) buckets 10 properties (
  "replication_num" = "3",
  "dynamic_partition.enable" = "true",
  "dynamic_partition.time_unit" = "DAY",
  "dynamic_partition.start" = "-7",
  "dynamic_partition.end" = "3",
  "dynamic_partition.prefix" = "par",
  "dynamic_partition.buckets" = "10",
  "dynamic_partition.hot_partition_num" = "1"
);
* */

    }
}
